import java.io.IOException;

import aws.EC2;

import pjaws.*;
import pjaws.paw.*;

import java.util.Random;

/**
 * PJAWS application example implementing a Vector Reduction with a master node.
 * Change TOTAL_NODES values to see how application scales.
 * 
 * @author Zachary Cancio
 * 
 */
public class VectorReductionExample extends Pjaws {
	public static final int TOTAL_NODES = 16; // TOTAL_NODES must be multiple of 2
	public static final int MASTER_ID = -1; //negative so there's not conflict with other nodes 

	// ---------CONTSRUCTORS-----------//
	public VectorReductionExample(String[] args) {
		super();

		// Setup EC2. Remember, only do this if the root option, "-r" is
		// specified
		if (args[0].equals("-r")) {
			String currDir = System.getProperty("user.dir");
			String key = currDir + "/pk-PEXNMTQQYNTJAPRC5H62B3KO7NO55QOV.pem";
			String cert = currDir
					+ "/cert-PEXNMTQQYNTJAPRC5H62B3KO7NO55QOV.pem";
			setEC2(new EC2(key, cert));
		}

		// create/setup nodes
		// determine iteration
		int iterations = TOTAL_NODES / 2 ;

		for (int i = 0; i < TOTAL_NODES; i++) {
			// instantiate new paw
			Node n = new Node(i, iterations);
			// add paw
			addPaw(n.getPawId(), n);
		}

		Master m = new Master(MASTER_ID, iterations);
		addPaw(MASTER_ID, m);

	}

	// ---------DEFINE PaWs----------//

	class Master extends Paw {
		private int myIterations;

		public Master(int id, int iterations) {
			super(id);
			myIterations = iterations;

			// define connections to the nodes
			for (int i = 0; i < TOTAL_NODES; i++) {
				PawSocket ps = new PawSocket(i);
				this.addPawSocket(ps);
			}
		}

		@Override
		public void run() {
			// receive each nodes initial minimum
			System.out.println("Initial Minimums");
			for (int id = 0; id < TOTAL_NODES; id++) {
				try {
					int min = (Integer) receivePawMessage(id, 5);
					System.out.println(id + "'s min: " + min);
				} catch (IOException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				} catch (ClassNotFoundException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}

			}

			for (int i = 0; i < myIterations; i++) {
				System.out.println("After Iteration " + i);
				for (int id = 0; id < TOTAL_NODES; id++) {
					try {
						int min = (Integer) receivePawMessage(id, 5);
						System.out.println(id + "'s min: " + min);
					} catch (IOException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					} catch (ClassNotFoundException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}

				}
			}

			System.out.println("Done!");
		}
	}

	class Node extends Paw {
		private int myLeftId;
		private int myRightId;
		private int myIterations;
		private int myMin;

		public Node(int id, int iterations) {
			super(id);
			myIterations = iterations;
			myMin = 0;

			// -----Add socket connection(s)
			// determine left node
			myLeftId = (id + TOTAL_NODES - 1) % TOTAL_NODES;
			PawSocket ps = new PawSocket(myLeftId);
			this.addPawSocket(ps);

			// determine right node
			myRightId = (id + TOTAL_NODES + 1) % TOTAL_NODES;
			ps = new PawSocket(myRightId);
			this.addPawSocket(ps);

			// Master
			ps = new PawSocket(MASTER_ID);
			this.addPawSocket(ps);

		}

		@Override
		public void run() {
			int pawId = this.getPawId();
			Random r = new Random();
			myMin = r.nextInt(); // pick a random number for starting min.

			System.out.println("Paw " + pawId + "'s min: " + myMin);

			// send master initial min
			try {
				sendPawMessage(myMin, MASTER_ID, 5);
			} catch (IOException e1) {
				// TODO Auto-generated catch block
				e1.printStackTrace();
			}

			int minLeft = myMin;
			int minRight = myMin;

			for (int i = 0; i < myIterations; i++) {
				System.out.println("Iteration: " + i);

				if (pawId % 2 == 0) { // my id is even then send first
					// (prevents deadlock)
					try {
						// Send! (order matters! Right then Left)
						sendPawMessage(myMin, myRightId, 1);
						sendPawMessage(myMin, myLeftId, 2);

						// Receive! (Left then Right)
						minLeft = (Integer) receivePawMessage(myLeftId, 1);
						minRight = (Integer) receivePawMessage(myRightId, 2);
					} catch (IOException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					} catch (ClassNotFoundException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
					System.out.println("");

				} else { // my id is odd then receive first (prevents
					// deadlock)

					try {
						// Receive! (Left then Right)
						minLeft = (Integer) receivePawMessage(myLeftId, 1);
						minRight = (Integer) receivePawMessage(myRightId, 2);

						// Send! (order matters! Right then Left)
						sendPawMessage(myMin, myRightId, 1);
						sendPawMessage(myMin, myLeftId, 2);

					} catch (IOException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					} catch (ClassNotFoundException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
				}

				// update minimum
				if (minLeft < myMin) {
					myMin = minLeft;
				}

				if (minRight < myMin) {
					myMin = minRight;
				}

				// send Master current min
				try {
					sendPawMessage(myMin, MASTER_ID, 5);
				} catch (IOException e1) {
					// TODO Auto-generated catch block
					e1.printStackTrace();
				}
			}

			System.out.println("FINAL MIN: " + myMin);
		}
	}

	public static void main(String[] args) {
		VectorReductionExample pjaws = new VectorReductionExample(args);
		pjaws.run(args);
		// int lo = (int) (Math.log(TOTAL_NODES) / Math.log(2));
		// System.out.print(lo);

	}

}
